package org.shj.hadoop;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.util.ReflectionUtils;

public class HadoopCodec {

	public static void main(String[] args) throws Exception{
		String inputFile = "e:\\hadoop\\wangmeishijie.txt";
		String outputFolder = "hdfs://192.168.56.101:9000/user/sj/";
		
		Configuration conf = new Configuration();
		
		//gzip
		long gzipTime = copyAndZipFile(conf, inputFile, outputFolder, 
				"org.apache.hadoop.io.compress.GzipCodec","gz");
		
		//bzip2
		long bzip2Time = copyAndZipFile(conf, inputFile, outputFolder, 
				"org.apache.hadoop.io.compress.BZip2Codec","bz2");
		
		//deflate
		long delfateTime = copyAndZipFile(conf, inputFile, outputFolder, 
				"org.apache.hadoop.io.compress.DefaultCodec","deflate");
		
		System.out.println("使用gzip压缩，时间为：" + gzipTime + "毫秒");
		System.out.println("使用bzip2压缩，时间为：" + bzip2Time + "毫秒");
		System.out.println("使用deflate压缩，时间为：" + delfateTime + "毫秒");
	}
	
	public static long copyAndZipFile(Configuration conf, String inputFile, String outputFolder,
			String codecClassName, String suffixName) throws Exception{
		long startTime = System.currentTimeMillis();
		InputStream in = new BufferedInputStream(new FileInputStream(inputFile));
		
		//去掉文件扩展名
		String baseName = inputFile.substring(inputFile.lastIndexOf("\\") + 1, inputFile.indexOf("."));
		String outputFile = outputFolder + baseName + "." + suffixName;
		
		FileSystem fs = FileSystem.get(URI.create(outputFile),conf, "shj");
		CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(Class.forName(codecClassName), conf);
		OutputStream out = codec.createOutputStream(fs.create(new Path(outputFile)));
		try{
			IOUtils.copyBytes(in, out, conf);
		}finally{
			IOUtils.closeStream(in);
			IOUtils.closeStream(out);
		}
		
		fs.close();
		
		long endTime = System.currentTimeMillis();
		return endTime - startTime;
	}
	
}
